package com.wuwangfu.transfor;

import com.wuwangfu.entity.DataBean;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Author jcshen
 * @Date 2023-02-22
 * @PackageName:com.wuwangfu.transfor
 * @ClassName: KeyByTransf
 * @Description:
 * @Version 1.0.0
 * <p>
 * https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/dev/datastream/operators/overview/#keyby
 */
public class KeyByTransf {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> line = env.socketTextStream("localhost", 8888);

        /*SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = line.map(w -> Tuple2.of(w, 1))
                .returns(Types.TUPLE(Types.STRING, Types.INT));
        SingleOutputStreamOperator<Tuple2<String, Integer>> summed = wordAndOne.keyBy(0)
                .sum(1);
        summed.print();*/

        //省份，城市，金额
        /**
         * a,b,100.1
         * a,b,200.1
         * a,c,200.1
         * a,c,100.1
         */
/*        SingleOutputStreamOperator<Tuple3<String, String, Double>> maped = line.map(new MapFunction<String, Tuple3<String, String, Double>>() {
            @Override
            public Tuple3<String, String, Double> map(String value) throws Exception {
                String[] fields = value.split(",");

                return Tuple3.of(fields[0], fields[1], Double.valueOf(fields[2]));
            }
        });
        maped.keyBy(0, 1)//根据 省份，城市 分组
                .sum(2)//聚合金额
                .print();*/


        SingleOutputStreamOperator<DataBean> maped = line.map(new MapFunction<String, DataBean>() {
            @Override
            public DataBean map(String value) throws Exception {
                String[] fields = value.split(",");
                DataBean bean = new DataBean();
                bean.setProvince(fields[0]);
                bean.setCity(fields[1]);
                bean.setMoney(Double.valueOf(fields[2]));
                return bean;
            }
        });
    /*    maped.keyBy("province","city")
                        .sum("money")
                                .print();*/

        /**
         * DataStream->KeyedStream->KeyGroupStreamPartitioner
         *          ->public int selectChannel(SerializationDelegate<StreamRecord<T>> record)
         *              KeyGroupRangeAssignment.assignKeyToParallelOperator(key, maxParallelism, numberOfChannels)
         *                  ->assignKeyToParallelOperator
         *                      computeOperatorIndexForKeyGroup(maxParallelism, parallelism, assignToKeyGroup(key, maxParallelism))
         *                          ->assignToKeyGroup
         *                              computeKeyGroupForKeyHash(key.hashCode(), maxParallelism);
         *
         *          ->computeOperatorIndexForKeyGroup
         *
         */

        maped.keyBy(new KeySelector<DataBean, DataBean>() {
            @Override
            public DataBean getKey(DataBean value) throws Exception {
                DataBean bean = new DataBean();
                bean.setProvince(value.getProvince());
                bean.setCity(value.getCity());
                return bean;
            }
        })
        .sum("money").print();


        env.execute();
    }
}
